package com.haiyou.common.spring.schedule;

import java.util.Collection;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import com.google.common.collect.Lists;
import com.haiyou.common.thread.policy.DefaultRejectedPolicy;
import com.haiyou.common.thread.task.AbstractOrderedCallback;
import com.haiyou.common.thread.task.AbstractOrderedTask;
import com.haiyou.common.util.collect.CollectionUtils;
import com.haiyou.common.util.hash.HashUtils;
import com.haiyou.common.util.string.StringUtils;
import com.haiyou.common.util.system.SystemUtils;

import lombok.extern.slf4j.Slf4j;

/**
 * 
 * 线程安全线程池 ： Actor模型+时效调度 绑定线程的任务池（模拟actor模型）
 * 线程间任务队列不可见，每个线程有自己的任务队列，可根据需求实现单线程模型的业务场景,任务不可插队，可执行时效任务
 * 
 * @Description: 
 * @author xingyuan
 * @date 2022年7月12日 下午6:11:06
 */
@Slf4j
final public class ActorScheduledExecutor implements ScheduledExecutorService {

	private final String threadNamedPrefix;
	private final ScheduledThreadPoolExecutor[] schedulers;

	public ActorScheduledExecutor(int nThreads, String prefix,RejectedExecutionHandler handler) {
		threadNamedPrefix = prefix;
		schedulers = new ScheduledThreadPoolExecutor[nThreads];
		for (int i = 0; i < schedulers.length; i++) {
			String n = StringUtils.merge(prefix, i);
			schedulers[i] = new ScheduledThreadPoolExecutor(1, (Runnable r) -> new Thread(r, n),
					handler);
		}
		log.warn("{} [{}] starting!!!", getClass().getSimpleName(), threadNamedPrefix);
	}
	
	public ActorScheduledExecutor(int nThreads, String prefix) {
		threadNamedPrefix = prefix;
		schedulers = new ScheduledThreadPoolExecutor[nThreads];
		for (int i = 0; i < schedulers.length; i++) {
			String n = StringUtils.merge(prefix, i);
			schedulers[i] = new ScheduledThreadPoolExecutor(1, (Runnable r) -> new Thread(r, n),
					DefaultRejectedPolicy.instance);
		}
		log.warn("{} [{}] starting!!!", getClass().getSimpleName(), threadNamedPrefix);
	}

	public void shutdownNowGracefully() {
		List<Runnable> l = shutdownNow();
		if (!CollectionUtils.isEmpty(l)) {
			Queue<Runnable> q = Lists.newLinkedList(l);
			while (q.size() > 0) {
				Runnable r = q.poll();
				r.run();
			}
		}
	}

	@Override
	public void shutdown() {
		for (int i = 0; i < schedulers.length; i++)
			schedulers[i].shutdown();
		log.warn("{} [{}] stopping!!!", getClass().getSimpleName(), threadNamedPrefix);
	}

	public void shutdownGracefully() {
		shutdown();
		while (!isTerminated())
			SystemUtils.sleep(1, TimeUnit.SECONDS);
		log.warn("{} [{}] stopped!!!", getClass().getSimpleName(), threadNamedPrefix);
	}

	@Override
	public List<Runnable> shutdownNow() {
		List<Runnable> list = Lists.newArrayList();
		for (int i = 0; i < schedulers.length; i++) {
			ScheduledExecutorService scheduledExecutorService = schedulers[i];
			List<Runnable> runnables = scheduledExecutorService.shutdownNow();
			list.addAll(runnables);
		}
		return list;
	}

	@Override
	public boolean isShutdown() {
		for (int i = 0; i < schedulers.length; i++) {
			ScheduledExecutorService scheduledExecutorService = schedulers[i];
			if (!scheduledExecutorService.isShutdown())
				return false;
		}
		return true;
	}

	@Override
	public boolean isTerminated() {
		for (int i = 0; i < schedulers.length; i++) {
			ScheduledExecutorService scheduledExecutorService = schedulers[i];
			if (!scheduledExecutorService.isTerminated())
				return false;
		}
		return true;
	}

	@Override
	public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
		for (int i = 0; i < schedulers.length; i++) {
			ScheduledExecutorService scheduledExecutorService = schedulers[i];
			scheduledExecutorService.awaitTermination(timeout, unit);
		}
		return true;
	}

	public ScheduledExecutorService getWorker(Object ser) {
		int index = 0;
		if (schedulers.length > 1) {
			index = HashUtils.hashCode(ser) % schedulers.length;
		}
		return schedulers[index];
	}

	@Override
	public <T> Future<T> submit(Callable<T> task) {
		AbstractOrderedCallback<?, ?> orderedCall = (AbstractOrderedCallback<?, ?>) task;
		ScheduledExecutorService worker = getWorker(orderedCall.getSer());
		Future<T> future = (Future<T>) worker.submit(orderedCall);
		return future;
	}

	@Override
	public <T> Future<T> submit(Runnable task, T result) {
		AbstractOrderedTask<?> orderedTask = (AbstractOrderedTask<?>) task;
		ScheduledExecutorService worker = getWorker(orderedTask.getSer());
		Future<T> future = (Future<T>) worker.submit(orderedTask, result);
		return future;
	}

	@Override
	public Future<?> submit(Runnable task) {
		AbstractOrderedTask<?> orderedTask = (AbstractOrderedTask<?>) task;
		ScheduledExecutorService worker = getWorker(orderedTask.getSer());
		Future<?> future = (Future<?>) worker.submit(orderedTask);
		return future;
	}

	@Override
	public void execute(Runnable command) {
		AbstractOrderedTask<?> orderedTask = (AbstractOrderedTask<?>) command;
		ScheduledExecutorService worker = getWorker(orderedTask.getSer());
		worker.execute(orderedTask);
	}

	@Override
	public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
		AbstractOrderedTask<?> orderedTask = (AbstractOrderedTask<?>) command;
		ScheduledExecutorService worker = getWorker(orderedTask.getSer());
		ScheduledFuture<?> scheduledFuture = (ScheduledFuture<?>) worker.schedule(orderedTask, delay, unit);
		return scheduledFuture;
	}

	@Override
	public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit) {
		AbstractOrderedCallback<?, ?> orderedCall = (AbstractOrderedCallback<?, ?>) callable;
		ScheduledExecutorService worker = getWorker(orderedCall.getSer());
		ScheduledFuture<V> scheduledFuture = (ScheduledFuture<V>) worker.schedule(orderedCall, delay, unit);
		return scheduledFuture;
	}

	@Override
	public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) {
		AbstractOrderedTask<?> orderedTask = (AbstractOrderedTask<?>) command;
		ScheduledExecutorService worker = getWorker(orderedTask.getSer());
		ScheduledFuture<?> scheduledFuture = (ScheduledFuture<?>) worker.scheduleAtFixedRate(orderedTask, initialDelay,
				period, unit);
		return scheduledFuture;
	}

	@Override
	public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit) {
		AbstractOrderedTask<?> orderedTask = (AbstractOrderedTask<?>) command;
		ScheduledExecutorService worker = getWorker(orderedTask.getSer());
		ScheduledFuture<?> scheduledFuture = (ScheduledFuture<?>) worker.scheduleWithFixedDelay(orderedTask,
				initialDelay, delay, unit);
		return scheduledFuture;
	}

	@Deprecated
	@Override
	public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException {
		throw new RuntimeException();
	}

	@Deprecated
	@Override
	public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
			throws InterruptedException {
		throw new RuntimeException();
	}

	@Deprecated
	@Override
	public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException {
		throw new RuntimeException();
	}

	@Deprecated
	@Override
	public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
			throws InterruptedException, ExecutionException, TimeoutException {
		throw new RuntimeException();
	}

}
